/*
 * StageTask.java
 */

package pipeline;


class StageTask<I,O> implements Runnable {

	
    private final Stage<I,O> stage;
    

    StageTask(Stage<I,O> stage) {
        this.stage     = stage;
    }
    
    
	public void run() {
		try {
			while (true) {
				
				/* get an item */
				stage.lock_get.lock();			// <GET>
				I in = stage.source.get();
				
				/* process it */
				stage.lock_process.lock();		// <PROCESS>
				stage.lock_get.unlock();		// </GET>
				O out = stage.processor.process(in);
				
				/* put result into sink */
				stage.lock_put.lock();			// <PUT>
				stage.lock_process.unlock();	// </PROCESS>
				stage.sink.put(out);
				
				/* notify stage listeners */
				stage.lock_notify.lock();		// <NOTIFY>
				stage.lock_put.unlock();		// </PUT>
				stage.notifyAll_ItemProcessed(out);
				
				stage.lock_notify.unlock();		// </NOTIFY>
			}
		}
	    catch (Exception exc) {
	    	//exc.printStackTrace();
	    }
	    finally {
	    	
	    	/* TODO: free resources better */
	    	if (stage.lock_get.isHeldByCurrentThread()) {
	    		stage.lock_get.unlock();
	    	}
	    	if (stage.lock_process.isHeldByCurrentThread()) {
	    		stage.lock_process.unlock();
	    	}
	    	if (stage.lock_put.isHeldByCurrentThread()) {
	    		stage.lock_put.unlock();
	    	}
	    	if (stage.lock_notify.isHeldByCurrentThread()) {
	    		stage.lock_notify.unlock();
	    	}
	    }
	}


}
